package com.bw.gmall.realtime.dim.app;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TableProcessDim;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.FlinkSourceUtil;
import com.bw.gmall.realtime.common.util.HbaseUtil;
import com.bw.gmall.realtime.dim.function.DimProcessFunction;
import com.bw.gmall.realtime.dim.function.DimSinkFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dim.app
 * @className: Test
 * @author: DuanYu
 * @description: 该类用于处理实时维度数据，将数据流经过 ETL 处理后，根据配置信息写入 HBase 中
 * @date: 2025/2/21 16:22
 * @version: 1.0
 */
public class DimApp extends BaseApp {
    public static void main(String[] args) {
        // 启动 DimApp 应用，传入主题名、应用名、并行度和其他相关参数
        new DimApp().start(Constant.TOPIC_DB, Constant.DIM_APP, 4, 10001);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 对输入的数据流进行 ETL 处理，过滤出符合条件的数据
        SingleOutputStreamOperator<JSONObject> getEtlStream = getGetEtlStream(dataStreamSource);
        // 从 MySQL 中获取配置表的变更数据，作为配置流
        DataStreamSource<String> processStream = env.fromSource(FlinkSourceUtil.getMysqlSource(Constant.PROCESS_DATABASE, Constant.PROCESS_DIM_TABLE_NAME), WatermarkStrategy.noWatermarks(), "cdc_strteam");
        // 对配置流进行处理，创建或删除 HBase 表，并将数据转换为 TableProcessDim 对象
        SingleOutputStreamOperator<TableProcessDim> createTableStream = createTable(processStream);
        // 定义一个广播状态，用于存储配置信息
        MapStateDescriptor<String, TableProcessDim> mapDescriptor = new MapStateDescriptor<>("broadcast_state", String.class, TableProcessDim.class);
        // 将配置流广播出去，以便所有并行任务都能获取到配置信息
        BroadcastStream<TableProcessDim> broadcast = createTableStream.broadcast(mapDescriptor);
        // 将 ETL 处理后的流与广播流连接，并使用 DimProcessFunction 进行处理
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> process = getEtlStream.connect(broadcast).process(new DimProcessFunction(mapDescriptor));
        // 对处理后的流进行过滤，只保留需要写入 HBase 的列
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> filterStream = getFilterStream(process);
        // 将过滤后的流写入 HBase 中
        filterStream.addSink(new DimSinkFunction());
    }

    /**
     * 对处理后的流进行过滤，只保留需要写入 HBase 的列
     * @param processBroadCastStream 处理后的流
     * @return 过滤后的流
     */
    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> getFilterStream(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDim>> processBroadCastStream) {
        return processBroadCastStream.map(new MapFunction<Tuple2<JSONObject, TableProcessDim>, Tuple2<JSONObject, TableProcessDim>>() {
            @Override
            public Tuple2<JSONObject, TableProcessDim> map(Tuple2<JSONObject, TableProcessDim> processDimTuple2) throws Exception {
                // 主流维度数据
                JSONObject f0 = processDimTuple2.f0;
                // 主流维度对应的配置表信息
                TableProcessDim f1 = processDimTuple2.f1;
                // 要写入 HBase 的列
                List<String> columns = Arrays.asList(f1.getSinkColumns().split(","));
                // 取出主流数据中的 data 部分
                JSONObject data = f0.getJSONObject("data");
                // 移除不需要写入 HBase 的列
                data.keySet().removeIf(key -> !columns.contains(key));
                return processDimTuple2;
            }
        });
    }

    /**
     * 处理配置流，根据配置信息创建或删除 HBase 表，并将数据转换为 TableProcessDim 对象
     * @param processStream 配置流
     * @return 转换后的 TableProcessDim 对象流
     */
    private static SingleOutputStreamOperator<TableProcessDim> createTable(DataStreamSource<String> processStream) {
        return processStream.flatMap(new RichFlatMapFunction<String, TableProcessDim>() {
            private Connection hbaseConnect;
            private TableProcessDim tableProcessDim;

            @Override
            public void open(Configuration parameters) throws Exception {
                hbaseConnect = HbaseUtil.getHbaseConnect();
            }

            @Override
            public void close() throws Exception {
                hbaseConnect.close();
            }

            @Override
            public void flatMap(String s, Collector<TableProcessDim> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                String op = jsonObject.getString("op");
                if ("c".equals(op) || "r".equals(op)) {
                    // 如果是创建或读取操作，将 after 部分转换为 TableProcessDim 对象
                    tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                    String[] split = tableProcessDim.getSinkFamily().split(",");
                    createTable(split);
                } else if ("d".equals(op)) {
                    // 如果是删除操作，将 before 部分转换为 TableProcessDim 对象
                    tableProcessDim = jsonObject.getObject("before", TableProcessDim.class);
                    deleteTable();
                } else if ("u".equals(op)){
                    // 如果是更新操作，将 after 部分转换为 TableProcessDim 对象
                    tableProcessDim = jsonObject.getObject("after", TableProcessDim.class);
                    String[] split = tableProcessDim.getSinkFamily().split(",");
                    // 先删除原表，再创建新表
                    deleteTable();
                    createTable(split);
                }
                // 设置操作类型
                tableProcessDim.setOp(op);
                // 将处理后的 TableProcessDim 对象输出
                collector.collect(tableProcessDim);
            }

            /**
             * 创建 HBase 表
             * @param families 表的列族
             */
            public void createTable(String[] families) {
                try {
                    System.out.println("tableProcessDim.getSinkTable() = " + tableProcessDim.getSinkTable());
                    // 调用 HBase 工具类创建表
                    HbaseUtil.createHBaseTable(hbaseConnect, Constant.HBASE_NAMESPACE, tableProcessDim.getSinkTable(), families);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            /**
             * 删除 HBase 表
             */
            public void deleteTable() {
                System.out.println("tableProcessDim.getSinkTable() = " + tableProcessDim.getSinkTable());
                try {
                    // 调用 HBase 工具类删除表
                    HbaseUtil.dropHBaseTable(hbaseConnect, Constant.HBASE_NAMESPACE, tableProcessDim.getSinkTable());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }).setParallelism(1);
    }

    /**
     * 对输入的数据流进行 ETL 处理，过滤出符合条件的数据
     * @param dataStreamSource 输入的数据流
     * @return ETL 处理后的流
     */
    private static SingleOutputStreamOperator<JSONObject> getGetEtlStream(DataStreamSource<String> dataStreamSource) {
        return dataStreamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    if (s != null) {
                        // 将输入的字符串解析为 JSON 对象
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        // 获取数据库名
                        String database = jsonObject.getString("database");
                        // 获取操作类型
                        String type = jsonObject.getString("type");
                        // 获取数据部分
                        String data = jsonObject.getString("data");
                        // 过滤出符合条件的数据
                        if (!"gmall".equals(database) && !"bootstrap-start".equals(type) && data != null && data.length() > 0) {
                            // 将符合条件的数据输出
                            collector.collect(jsonObject);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }
}